package com.distributer.delayqueue.delayqueue.bucket;

import java.util.List;
import java.util.TimerTask;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONArray;
import com.distributer.delayqueue.delayqueue.JobOperationService;
import com.distributer.delayqueue.delayqueue.event.EventBus;
import com.distributer.delayqueue.delayqueue.event.RedisJobTraceEvent;
import com.distributer.delayqueue.delayqueue.support.DistributedLock;
import com.distributer.delayqueue.delayqueue.support.RedisQueueProperties;
import com.distributer.delayqueue.job.JobMsg;
import com.distributer.delayqueue.job.JobStatus;

public class BucketTask extends TimerTask{

	private static final Logger LOGGER = LoggerFactory.getLogger(BucketTask.class);
	 public static final long TIME_OUT = 1000 * 30;
	private static final int DEFAUT_PEEK_JOB_SIZE = 10;
	private JobOperationService jobOperationService;
	private String bucketName;
	private String readyName;
	private RedisQueueProperties redisQueueProperties;
	private DistributedLock lock = null;
	
	public BucketTask(String bucketName){
		this.bucketName = bucketName;
	}
	
	@Override
	public void run() {
		runTemplate();
	}
	
	public void runTemplate(){
		if (redisQueueProperties.isCluster()) {
			try{
				lock.lock(bucketName);
				runInstance();
			}finally{
				lock.unlock(bucketName);
			}
		}else {
			runInstance();
		}
	}
	
	public void runInstance(){
		//LOGGER.info(String.format("开始轮询...%s", bucketName));
		List<String> jobsIds = peeks();
		if (jobsIds !=null && jobsIds.size()>0) {
			//LOGGER.info(String.format("延时队列: %s 获取任务列表: %s", bucketName,JSONArray.toJSONString(jobsIds)));
			for (String jobId : jobsIds) {
				if (!StringUtils.isEmpty(jobId)) {
					//查询jobId是否存在
					JobMsg jobMsg = jobOperationService.getJob(jobId);
					if (jobMsg == null) {
						//从bucket移除jobId
						jobOperationService.removeBucketKey(bucketName, jobId);
						continue;
					}
					try{
						if (lock.lock(jobMsg.getJobId())) {
							//判断job状态是否已经删除
							if (jobMsg.getStatus() == JobStatus.Delete.ordinal()) {
								jobOperationService.removeJobToPool(jobId);
								jobOperationService.removeBucketKey(bucketName, jobId);
								continue;
							}
							//判断是否是延时任务
							if (jobMsg.getStatus() != JobStatus.Delay.ordinal()) {
								continue;
							}
							//判断是否已经达到延时执行时间
							int result =(int) (((jobMsg.getCreateTime().getTime()) + jobMsg.getDelayTime()) - System.currentTimeMillis());
							if (result <= 0) {
								//到达执行时间的延时任务，将任务加到实时队列中
								jobMsg.setStatus(JobStatus.Ready.ordinal());
								//修改元数据状态,避免timer竞争到数据导致数据不一致
								EventBus.getInstance().postEvent(new RedisJobTraceEvent(jobMsg));
								//添加任务到实时队列
								jobOperationService.addReadyTime(readyName, jobId);
								//从延时队列中删除任务
								jobOperationService.removeBucketKey(bucketName, jobId);
								//LOGGER.info("任务id:{}到执行时间!",jobId);
							}
						}
					}catch(Exception e){
						LOGGER.error("添加job到实时队列失败！",e);
					}finally{
						lock.unlock(jobMsg.getJobId());
					}
				}
			}
		}
	}

	/**
	 * 批量获取任务
	 * @return
	 */
	private List<String> peeks(){
		List<String> topJobs = jobOperationService.getBucketTopJobs(bucketName, DEFAUT_PEEK_JOB_SIZE);
		return topJobs;
	}
	
	public void setJobOperationService(JobOperationService jobOperationService) {
		this.jobOperationService = jobOperationService;
	}

	public void setBucketName(String bucketName) {
		this.bucketName = bucketName;
	}

	public void setReadyName(String readyName) {
		this.readyName = readyName;
	}

	public void setRedisQueueProperties(RedisQueueProperties redisQueueProperties) {
		this.redisQueueProperties = redisQueueProperties;
	}

	public void setLock(DistributedLock lock) {
		this.lock = lock;
	}

}
